package com.cetccloud.ap.service.impl;

import com.cetccloud.ap.service.ProxyService;
import com.cetccloud.ap.util.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.util.Arrays;

/**
 * @ClassName ProxyServiceImpl
 * @Description TODO
 * @Author superz
 * @Date 2021/4/21 16:10
 * @Version 1.0
 **/
@Slf4j
@Service
public class ProxyServiceImpl implements ProxyService {

    @Autowired(required = false)
    private AdminClient adminClient;

    @Override
    public boolean createTopic(String topic) {
        try {
            NewTopic _topic = new NewTopic(topic, 3, (short) 1);
            CreateTopicsResult result = adminClient.createTopics(Arrays.asList(_topic));
            result.all().get();
            log.info("创建topic: {}", topic);
            return true;
        } catch (Exception e) {
            log.error("createTopic error", e);
            return false;
        }
    }

    @Override
    public Mono _createTopic(String topic) {
        try {
            NewTopic _topic = new NewTopic(topic, 3, (short) 1);
            CreateTopicsResult result = adminClient.createTopics(Arrays.asList(_topic));
            return Mono.create(monoSink -> {
                result.all().whenComplete((v, t) -> {
                    if (null == t) {
                        monoSink.success(Constants.MONO_SUCCESS);
                    } else {
                        monoSink.error(t);
                    }
                });
            });
        } catch (Exception e) {
            log.error("createTopic error", e);
            return Mono.error(e);
        }
    }
}
